package com.chukun.multi.masterslave.reusable;

import com.chukun.multi.twophase.AbstractTerminatableThread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

/**
 * 基于工作者线程的slave的参与者的通用实现
 * @param <T> 子任务的类型
 * @param <V> 子任务的处理结果的类型
 */
public abstract class WorkerThreadSlave<T,V> extends AbstractTerminatableThread implements SlaveSpec<T,V> {

    private final BlockingQueue<Runnable> taskQueue;

    public WorkerThreadSlave(BlockingQueue<Runnable> taskQueue) {
        this.taskQueue = taskQueue;
    }

    /**
     * 提交任务
     * @param task 子任务
     * @return
     * @throws InterruptedException
     */
    public Future<V> submit(final T task) throws InterruptedException{
        FutureTask<V> ft = new FutureTask<>(() -> {
            V result;
            try{
                result = doProcess(task);
            }catch (Exception e) {
               SubTaskFailureException stfe = newSubTaskFailureException(task,e);
               throw stfe;
            }
            return result;
        });
        taskQueue.put(ft);
        terminationToken.reservations.incrementAndGet();
        return ft;
    }

    /**
     * 生成重试任务的异常
     * @param task
     * @param cause
     * @return
     */
    private SubTaskFailureException newSubTaskFailureException(final T task,Exception cause) {
        RetryInfo retryInfo = new RetryInfo(task, () -> {
            V result;
            result = doProcess(task);
            return result;
        });
        return new SubTaskFailureException(retryInfo,cause);
    }

    /**
     * 留给子类实现，实现任务的处理逻辑
     * @param task
     * @return
     * @throws Exception
     */
    protected abstract V doProcess(T task) throws Exception;

    @Override
    protected void doRun() throws Exception {
        try{
            Runnable take = taskQueue.take();
            take.run();
        }finally {
            terminationToken.reservations.decrementAndGet();
        }
    }

    @Override
    public void init() {
        start();
    }

    @Override
    public void shutdown() {
        terminate();
    }
}
